/// \file		WnTCPServer.cpp
/// \brief		WIND通信(TCP)服务端类的定义文件
/// \author		xuefeng
/// \version	1.0
/// \date		2011-07-19
/// \history	2011-07-19 新建

#include "ComHeader.h"
#include "WnTCPServer.h"
#include "WnBaseIOService.h"
#include "WnUtility.h"
#include "ITCPServer.h"
using namespace boost::system;

const auto MAX_CLIENT_REF_COUNTER = 2;

// 读写锁
typedef boost::shared_lock<boost::shared_mutex> readLock;
typedef boost:: unique_lock<boost::shared_mutex> writeLock;

enum ENU_RUN_STATE
{
  E_WAIT,
  E_RUN,
  E_STOP,
};

WnTCPServer::WnTCPServer(int id, ITCPServerNotify *poNotify, int HBInterval, unsigned short usThreadCount)
  : WnEndPoint(id)
  , m_apAcceptor()
  , m_apListenThread()
  , m_nRunState(E_WAIT)
  , m_oClientsMap()
  , m_mtx_ClientsMap()
  , m_oOperateMutex()
  , m_poNotify(poNotify)
  , m_oNotifyMutex()
  , m_HB_interval(HBInterval)
{
  m_poBaseIOService = new WnBaseIOService();
  m_poBaseIOService->StartService(usThreadCount);
}

WnTCPServer::~WnTCPServer(void)
{
  // 关闭socket
  CloseSocket();

  // 停止ios服务
  m_poBaseIOService->StopService();
  delete m_poBaseIOService;
}

bool WnTCPServer::ServerListen(unsigned short usPort, bool blAuth)
{
  return StartListen(usPort, blAuth);
}

bool WnTCPServer::SendDataToClient(std::string clientID, const unsigned char *pbData, unsigned int unDataSize)
{
  return SendData(clientID, pbData, unDataSize);
}

void WnTCPServer::CloseSocket()
{
  boost::mutex::scoped_lock oLock(m_oOperateMutex);

  m_nRunState = E_STOP;

  if(m_apListenThread)
  {
    m_apListenThread->join();
    m_apListenThread.reset();
  }

  if(m_apAcceptor)
  {
    m_apAcceptor->Stop();
    m_apAcceptor.reset();
  }

  {
    writeLock oMapLock(m_mtx_ClientsMap);
    for(auto it = m_oClientsMap.begin(); it != m_oClientsMap.end(); ++it)
    {
      it->second->apHeartBeatTimer->cancel();
      it->second->apTimeOutTimer->cancel();
      it->second->apConnection->Close();
    }
    m_oClientsMap.clear();
  }
}

void WnTCPServer::OnAccept(CONNECTION_APTR apConnection)
{
  CLIENT_ITEM_APTR apClientItem(new TAG_CLIENT_ITEM);
  apClientItem->apConnection = apConnection;
  apClientItem->apHeartBeatTimer.reset(new deadline_timer(m_poBaseIOService->GetIOService()));
  apClientItem->apTimeOutTimer.reset(new deadline_timer(m_poBaseIOService->GetIOService()));
  apClientItem->SetCheckIndex(0);

  {
    // MDF方式下，启动心跳
    apClientItem->apHeartBeatTimer->expires_from_now(millisec(m_HB_interval));
    apClientItem->apHeartBeatTimer->async_wait(boost::bind(&WnTCPServer::OnHeartBeatTimer, this, apClientItem, boost::asio::placeholders::error));

    // MDF方式下，启动超时定时器
    apClientItem->apTimeOutTimer->expires_from_now(millisec(m_HB_interval * NO_HEARTBEAT_COUNT));
    apClientItem->apTimeOutTimer->async_wait(boost::bind(&WnTCPServer::OnNoRecvDataTimer, this, apClientItem->apConnection, boost::asio::placeholders::error));
  }

  {
    writeLock oMapLock(m_mtx_ClientsMap);
    m_oClientsMap.insert(std::pair<std::string, CLIENT_ITEM_APTR>(apConnection->clientID(), apClientItem));
  }

  boost::thread t([&,apConnection]()
  {
    WnConnection<boost::asio::ip::tcp> *poConnection = dynamic_cast<WnConnection<boost::asio::ip::tcp>*>(apConnection.get());
    if(poConnection != NULL)
    {
      WnConnection<boost::asio::ip::tcp>::endpoint_type oEndPoint = poConnection->GetRemoteEndPoint();
      readLock lock(m_oNotifyMutex);
      if(m_poNotify != NULL)
        m_poNotify->OnAccept(apConnection->clientID(), oEndPoint.address().to_string().c_str(), oEndPoint.port());
    }
  });
}

void WnTCPServer::OnDisconnect(CONNECTION_APTR apConnection, int nStatus)
{
	CLIENT_ITEM_APTR client_ptr = nullptr;
	{
		writeLock oMapLock(m_mtx_ClientsMap);
		auto it = m_oClientsMap.find(apConnection->clientID());
		if (it != m_oClientsMap.end())
		{
			client_ptr = it->second;
			it = m_oClientsMap.erase(it);
		}
	}

	if (client_ptr)
	{
		client_ptr->apHeartBeatTimer->cancel();
		client_ptr->apTimeOutTimer->cancel();

		boost::thread t([&, apConnection, nStatus]()
		{
			readLock lock(m_oNotifyMutex);
			if (m_poNotify != NULL)
				m_poNotify->OnClientDisconnect(apConnection->clientID());
		});
	}
}

void WnTCPServer::OnReceiveData(CONNECTION_APTR apConnection, const PACKET_APTR& apPacket)
{
	CLIENT_ITEM_APTR client_ptr = nullptr;
	{
		readLock oMapLock(m_mtx_ClientsMap);
		auto it = m_oClientsMap.find(apConnection->clientID());
		if (it != m_oClientsMap.end())
			client_ptr = it->second;
	}

	if (client_ptr)
	{
		// 取消原定时器
		client_ptr->apTimeOutTimer->cancel();
		// 重启定时器
		client_ptr->apTimeOutTimer->expires_from_now(millisec(m_HB_interval * NO_HEARTBEAT_COUNT));
		client_ptr->apTimeOutTimer->async_wait(boost::bind(&WnTCPServer::OnNoRecvDataTimer, this, 
			client_ptr->apConnection, boost::asio::placeholders::error));

		if ((apPacket->GetPacketData() != NULL) && !apPacket->IsHeartBeatPacket())
		{
			readLock lock(m_oNotifyMutex);
			if (m_poNotify != NULL)
			{
				m_poNotify->OnRecvClientData(apConnection->clientID(), apPacket->GetBodyData(), apPacket->GetBodyLength());
			}
		}
	}
}

void WnTCPServer::OnDataError(CONNECTION_APTR apConnection)
{
	CLIENT_ITEM_APTR client_ptr = nullptr;
	{
		readLock oMapLock(m_mtx_ClientsMap);
		auto it = m_oClientsMap.find(apConnection->clientID());
		if (it != m_oClientsMap.end())
			client_ptr = it->second;
	}

	if (client_ptr)
	{
		// 取消原定时器
		client_ptr->apTimeOutTimer->cancel();
		// 重启定时器
		client_ptr->apTimeOutTimer->expires_from_now(millisec(m_HB_interval * NO_HEARTBEAT_COUNT));
		client_ptr->apTimeOutTimer->async_wait(boost::bind(&WnTCPServer::OnNoRecvDataTimer, this, 
			client_ptr->apConnection, boost::asio::placeholders::error));

		boost::thread t([&, apConnection]()
		{
			readLock lock(m_oNotifyMutex);
			if (m_poNotify != NULL)
				m_poNotify->OnClientDataError(apConnection->clientID());
		});
	}
}

void WnTCPServer::ClearNotifyObj()
{
  writeLock lock(m_oNotifyMutex);
  m_poNotify = NULL;
}

void WnTCPServer::Close()
{
  CloseSocket();
}

bool WnTCPServer::StartListen(unsigned short usPort, bool blAuth)
{
  bool blResult = false;
  CloseSocket();

  boost::mutex::scoped_lock oLock(m_oOperateMutex);
  m_apAcceptor.reset(new WnAcceptor<boost::asio::ip::tcp>( shared_from_this(), m_poBaseIOService->GetIOService()));
  if(0 != m_apAcceptor->SetAddr("0.0.0.0", boost::lexical_cast<string>(usPort)))
  {
    // 发送通知
    {
      stringstream ss;
      ss << "file:" << __FILE__ << ", function:" << __FUNCTION__ << ", line:" << __LINE__ << ". @@@. "
        << "设置服务器端地址及端口错误，端口疑似被占用. port:" << usPort;

      OnMsg(2, ss.str());
    }
    return blResult;
  }


  m_nRunState = E_WAIT;

  m_apListenThread.reset(new boost::thread(boost::bind(&WnTCPServer::ListenThread, this)));
  while(m_nRunState == E_WAIT)
    boost::this_thread::sleep(millisec(NORMAL_SLEEP_TIME));

  if(m_nRunState == E_RUN)
    blResult = true;
  else
  {
    m_apListenThread->join();
    m_apListenThread.reset();
  }
  return blResult;
}

bool WnTCPServer::SendData(std::string clientID, const unsigned char *pbData, unsigned int unDataSize)
{
  bool blResult = false;

  if(pbData == NULL || unDataSize == 0) // 发送数据有误
    return blResult;

  CONNECTION_APTR apConnection;
  {
    readLock oMapLock(m_mtx_ClientsMap);
    auto it = m_oClientsMap.find(clientID);
    if((it == m_oClientsMap.end()) || !it->second->apConnection) // session无效
      return blResult;

    apConnection = it->second->apConnection;
  }

  // 发送数据
  PACKET_APTR apPacket;
  apPacket.reset(new WnPacket(pbData, unDataSize, 0, 0));
  ENU_ASYNSEND_RESULT eResult = apConnection->AsyncSend(apPacket, false);

  // 判断发送结果
  if(eResult != E_AR_SUCCESS)
  {
    if((eResult == E_AR_NOSEND) || (eResult == E_AR_OVERFLOW) || (eResult == E_AR_NOCONNECT))
    {
      // 关闭连接
      apConnection->Close();

      // 发出通知
      OnDisconnect(apConnection, E_DISCONNECT_IOERROR);
    }
  }

  return eResult == E_AR_SUCCESS;
}

void WnTCPServer::ListenThread()
{
  ASSERT(m_apAcceptor);

  if(m_apAcceptor->Start())
    m_nRunState = E_RUN;
  else
    m_nRunState = E_STOP;

  while(m_nRunState == E_RUN)
    boost::this_thread::sleep(millisec(NORMAL_SLEEP_TIME));
}

void WnTCPServer::OnHeartBeatTimer(CLIENT_ITEM_APTR apClientItem, const boost::system::error_code& oErrorCode)
{
  if(oErrorCode)
    return;

  // 如果连接已经断开，则关闭定时器
  if(!apClientItem->apConnection->IsConnected())
    return;

  apClientItem->apHeartBeatTimer->expires_from_now(millisec(m_HB_interval));
  apClientItem->apHeartBeatTimer->async_wait(boost::bind(&WnTCPServer::OnHeartBeatTimer, this, apClientItem, boost::asio::placeholders::error));

  // 获取时间值
  uint64_t ullTime = WnUtility::GetTimeValue();

  // 字节转换
  unsigned char bStream[sizeof(ullTime)];
  WnUtility::IntegerToStream(ullTime, bStream);

  // 发送心跳包
  apClientItem->apConnection->AsyncSend(PACKET_APTR(new WnPacket(bStream, sizeof(ullTime), apClientItem->GetCheckIndex(), HEARTBEAT_BITFLAG)), true);
}

void WnTCPServer::OnNoRecvDataTimer(CONNECTION_APTR apConnection, const boost::system::error_code& oErrorCode)
{
  if(oErrorCode)
    return;

  // 关闭连接
  apConnection->Close();

  // 发出通知
  OnDisconnect(apConnection, E_DISCONNECT_RCVTIMEOUT);
}

void WnTCPServer::OnMsg(int type, const std::string& msg)
{
  readLock lock(m_oNotifyMutex);
  if(m_poNotify)
    m_poNotify->OnTCPMsg(type, msg);
}

void WnTCPServer::RegisterSpi(ITCPServerNotify *pSpi)
{
  writeLock lock(m_oNotifyMutex);
  m_poNotify = pSpi;
}

bool WnTCPServer::OtherServerListen(unsigned short usPort, bool blAuth /*= false*/)
{
  return false;
}

bool WnTCPServer::SendDataToOtherClient(std::string sessionID, const unsigned char *pbData, unsigned int unDataSize)
{
  return false;
}

extern "C"
{
  boost::mutex gMutex_Server;

  /// \brief 通信终端表
  LIST_END_POINT gEndPointList_Server;

  int gID_Server = 0;

  ITCPServer* CreateTCPServerInstance(ITCPServerNotify* poNotifyObj, int HBInterval, unsigned short usThreadCount)
  {
    boost::mutex::scoped_lock oGlobalLock(gMutex_Server);
    WnTCPServer* pObj = new WnTCPServer(++gID_Server, poNotifyObj, HBInterval, usThreadCount);
    gEndPointList_Server.push_back(END_POINT_APTR(pObj));
    return pObj;
  }

  void ReleaseTCPServerInstance(ITCPServer *poServerInstance)
  {
    boost::mutex::scoped_lock oGlobalLock(gMutex_Server);

    poServerInstance->CloseSocket();
    for(auto it = gEndPointList_Server.begin(); it != gEndPointList_Server.end(); ++it)
    {
      if((*it)->GetID() == static_cast<WnTCPServer*>(poServerInstance)->GetID())
      {
        gEndPointList_Server.erase(it);
        break;
      }
    }
  }
}